package task

import (
	"context"
	"fmt"
	"log/slog"
	"runtime"
	"sync"
	"sync/atomic"
	"time"
)

const (
	minShards          = 1                     // 任务队列存储分片数量最小值 1
	maxShards          = 512                   // 任务队列存储分片数量最大值 512
	defaultShards      = 32                    // 任务队列存储分片数量默认值 32 (表示创建32个定任务工作队列)
	minWorkers         = 1                     // 任务并发执行协程最小数量 1
	maxWorkers         = 262144                // 任务并发执行协程最大数量 262144, 262144*4KB = 1048576KB = 1024MB,  任务协程至少需消耗 1024MB 内存
	defaultWorkers     = 2048                  // 任务并发执行协程默认数量, 2048*4KB = 8192KB = 8MB, 任务执行协程至少需消耗 8M 内存
	minCheckPeriod     = time.Millisecond      // 任务队列检查最小时间最小值 1毫秒
	maxCheckPeriod     = time.Second           // 任务队列检查最小时间最大值 1秒
	defaultCheckPeriod = 10 * time.Millisecond // 任务队列检查最小时间默认值 10 毫秒，决定了定时任务定时/间隔精度
	defaultTimeout     = 30 * time.Second      // 默认任务执行超时时间为30秒
)

type Options struct {
	ShardCount  int           // 任务队列存储分片数量,  默认 32, 最小 1, 最大 512
	WorkerCount int           // 任务处理协程数量, 默认 2048, 最小 1, 最大 262144
	CheckPeriod time.Duration // 任务队列检查最小时间，决定了定时任务定时/间隔精度， 默认 10 毫秒, 最小 1毫秒, 最大 1000毫秒(1秒)
}

var defaultOptions = Options{
	ShardCount:  defaultShards,
	WorkerCount: defaultWorkers,
	CheckPeriod: defaultCheckPeriod,
}

// TaskManager 定时任务管理器
type TaskManager struct {
	name       string
	options    *Options
	scheduleds []*taskScheduled // 定时任务队列调度器分片组(依据任务Key进行FNV32分片)
	monitor    *taskMonitor     // 任务运行监控器
	isGlobal   bool             // 是否全局默认任务管理器
	wg         *sync.WaitGroup
	startCh    chan struct{} // 启动信号
	stopCh     chan struct{} // 停止信号
	workerCh   chan *Task    // 任务执行协程工作信号, 用于限制任务并发执行协程数量
	logCtx     context.Context
	isRuning   *atomic.Bool
}

// NewTaskManager 创建一个新的定时任务管理器
//   - name string   管理器名称
//   - options *Options  配置选项
//   - type Options struct {
//   - ShardCount  int           // 任务队列存储分片数量,  默认 32, 最小 1, 最大 512
//   - WorkerCount int           // 任务处理协程数量, 默认 2048, 最小 1, 最大 262144
//   - CheckPeriod time.Duration // 任务队列检查最小时间，决定了定时任务定时/间隔精度， 默认 10 毫秒, 最小 1毫秒, 最大 1000毫秒(1秒)
//   - }
func NewTaskManager(name string, options *Options) *TaskManager {
	opts := checkOptions(options)
	tm := &TaskManager{
		name:       name,
		options:    opts,
		scheduleds: make([]*taskScheduled, opts.ShardCount),
		monitor:    newTaskMonitor(opts.WorkerCount),
		wg:         new(sync.WaitGroup),
		startCh:    make(chan struct{}),
		stopCh:     make(chan struct{}),
		workerCh:   make(chan *Task, opts.WorkerCount),
		logCtx:     context.Background(),
		isRuning:   new(atomic.Bool),
	}
	// 初始化分片
	for i := range opts.ShardCount {
		tm.scheduleds[i] = &taskScheduled{
			taskManager: tm,
			taskQueue:   newTaskQueue(),
			taskMaps:    newTaskMaps(),
			timer:       nil,
			nextTimeMs:  new(atomic.Int64),
			isRuning:    new(atomic.Bool),
		}
	}
	go tm.run(tm.getEntryPC(3, true))
	// 等待启动完成信号
	<-tm.startCh
	return tm
}

// checkOptions 检查配置选项
func checkOptions(options *Options) *Options {
	if options == nil {
		return &defaultOptions
	}
	opts := *options
	opts.ShardCount = checkOptionMinMax(opts.ShardCount, defaultShards, minShards, maxShards)
	opts.WorkerCount = checkOptionMinMax(opts.WorkerCount, defaultWorkers, minWorkers, maxWorkers)
	opts.CheckPeriod = checkOptionMinMax(opts.CheckPeriod, defaultCheckPeriod, minCheckPeriod, maxCheckPeriod)
	return &opts
}

// checkOptionMinMax 检查选项值是否在最小最大范围内，不在则返回默认值或范围边界值
func checkOptionMinMax[T int | time.Duration](val, defaultVal, min, max T) T {
	switch {
	case val == 0:
		return defaultVal
	case val < min:
		return min
	case val > max:
		return max
	default:
		return val
	}
}

// run 运行定时任务管理器，并等待所有分片执行器启动完成，监听stopCh停止信号

func (ts *TaskManager) run(pc uintptr) {
	ts.logFormat(
		pc, slog.LevelInfo,
		"run(shard/worker:%d)...", ts.options.ShardCount)
	var startWG sync.WaitGroup
	// 启动任务队列分片执行器
	ts.wg.Add(ts.options.ShardCount)
	startWG.Add(ts.options.ShardCount)
	for i := range ts.options.ShardCount {
		go func(i int) {
			defer ts.wg.Done()
			scheduled := ts.scheduleds[i]
			scheduled.runWorker()
			scheduled.runQueue()
			startWG.Done()
			// 等待停止信号
			<-ts.stopCh
			scheduled.isRuning.Store(false)
		}(i)
	}
	startWG.Wait()
	// 启动任务运行监控器
	ts.wg.Add(1)
	startWG.Add(1)
	go func() {
		defer ts.wg.Done()
		ts.monitor.Run()
		startWG.Done()
		// 等待停止信号
		<-ts.stopCh
		ts.logFormat(
			pc, slog.LevelInfo,
			"stop(shard/worker:%d)...", ts.options.ShardCount)
		ts.monitor.Stop()
	}()

	startWG.Wait()
	// 通知启动完成
	ts.isRuning.Store(true)
	ts.startCh <- struct{}{}
}

// getScheduled 获取任务队列调度器分片
func (ts *TaskManager) getScheduled(taskKey string) *taskScheduled {
	if ts.options.ShardCount == 1 {
		return ts.scheduleds[0]
	}
	return ts.scheduleds[fnv32String(taskKey)%uint32(ts.options.ShardCount)]
}

// addTask 添加任务
func (ts *TaskManager) addTask(task *Task) error {
	if !ts.isRuning.Load() {
		return ErrTaskClosed
	}
	if task == nil {
		return ErrTaskIsNil
	}
	if task.key == "" {
		return ErrTaskKeyEmpty
	}
	if task.handler == nil {
		return ErrTaskHandlerNil
	}
	if !task.isOnce {
		// 周期任务
		if task.interval == nil {
			return ErrTaskInterval
		}
		task.executeAt = task.interval.ExecuteAt()
	}
	if task.executeAt.IsZero() {
		return ErrTaskExecuteAt
	}
	task.scheduled = ts.getScheduled(task.key)
	if oldTask, loaded := task.scheduled.taskQueue.LoadOrPush(task); loaded {
		// 一次性延迟缓存任务，忽略重复添加
		if oldTask.isBuffer {
			return nil
		}
		// 其他任务, 重复添加则先删除队列中的旧任务
		task.scheduled.taskQueue.Remove(task.key)
	} else {
		// 添加到任务信息存储器后, 更新监控信息
		ts.monitor.IncAdded()
	}
	// 添加到任务信息到全部映射信息
	task.scheduled.taskMaps.Store(task.key, &TaskInfo{
		Key:      task.key,
		IsOnce:   task.isOnce,
		IsBuffer: task.isBuffer,
		RunMax:   task.runMax,
		Timeout:  task.timeout,
	})
	// 添加到分片队列调度器中
	task.scheduled.taskQueue.Push(task)
	// 通知调度器
	task.scheduled.timerSignal(task)
	return nil
}

// updateIntervalTask 更新任务信息
func (ts *TaskManager) updateIntervalTask(task *Task) error {
	if task == nil {
		return ErrTaskIsNil
	}
	if task.isOnce {
		return nil
	}
	task.runCount++
	isNext := task.runMax == 0 || (task.runMax > 0 && task.runCount < task.runMax)
	if !isNext {
		return nil
	}
	task.executeAt = task.interval.ExecuteAt()
	task.state = TaskStatePending
	// 添加任务到分片队列调度器中
	task.scheduled.taskQueue.Push(task)
	// 通知调度器
	task.scheduled.timerSignal(task)
	return nil
}

// RemoveTask 移除指定任务
func (ts *TaskManager) RemoveTask(keys ...string) error {
	for _, taskKey := range keys {
		if taskKey == "" {
			continue
		}
		scheduled := ts.getScheduled(taskKey)
		scheduled.taskMaps.Delete(taskKey)
		task, ok := scheduled.taskQueue.LoadAndRemove(taskKey)
		if ok {
			// 原子更新状态
			if !atomic.CompareAndSwapInt32(&task.state, TaskStatePending, TaskStateRemoved) {
				atomic.CompareAndSwapInt32(&task.state, TaskStateRunning, TaskStateRemoved)
			}
		}
	}
	return nil
}

// RemoveAllTasks 删除所有任务
func (ts *TaskManager) RemoveAllTasks() error {
	for _, scheduled := range ts.scheduleds {
		scheduled.taskQueue.Reset()
		scheduled.taskMaps.Reset()
	}
	return nil
}

// GetTask 获取指定key的定时任务信息
func (ts *TaskManager) GetTask(key string) (TaskInfo, error) {
	task, ok := ts.getScheduled(key).taskMaps.Load(key)
	if !ok {
		return TaskInfo{}, ErrTaskKeyNotExist
	}
	return *task, nil
}

// GetAllTasks 获取所有定时任务信息
func (ts *TaskManager) GetAllTasks() []TaskInfo {
	tasks := make([]TaskInfo, 0)
	for _, scheduled := range ts.scheduleds {
		data := scheduled.taskMaps.Data()
		for _, task := range data {
			tasks = append(tasks, *task)
		}
	}
	return tasks
}

// Size 获取定时任务数量(包含未运行、已运行的全部任务)
//   - 包含全部定时任务的映射(可主动通过RemoveTask删除指定任务信息)
func (ts *TaskManager) Size() int {
	c := 0
	for _, scheduled := range ts.scheduleds {
		c += scheduled.taskMaps.Size()
	}
	return c
}

// IsRuning 获取定时任务管理器运行状态
func (ts *TaskManager) IsRuning() bool {
	return ts.isRuning.Load()
}

// Monitor 获取定时任务运行监控信息
func (ts *TaskManager) Monitor() TaskMonitorData {
	return ts.monitor.Data()
}

// Stop 关闭定时任务管理器
//   - 注意：仅停止定时任务管理器的运行，不会删除任务数据
func (ts *TaskManager) Stop() {
	if ts.isRuning.CompareAndSwap(true, false) {
		close(ts.stopCh)
		close(ts.startCh)
		close(ts.workerCh)
		ts.wg.Wait()
	}
}

// Restart 重启定时任务管理器
//   - 注意：重启后，所有到期任务会立即运行
func (ts *TaskManager) Restart() {
	if ts.isRuning.CompareAndSwap(false, true) {
		ts.stopCh = make(chan struct{})
		ts.startCh = make(chan struct{})
		ts.workerCh = make(chan *Task, ts.options.WorkerCount)
		ts.wg = new(sync.WaitGroup)
		pc := ts.getEntryPC(3, true)
		ts.log(
			pc, slog.LevelInfo,
			"restart...")
		ts.monitor.Stop()
		go ts.run(pc)
		// 等待时间轮启动完成信号
		<-ts.startCh
	}
}

// getEntryPC 获取EventBus 的调用栈入口PC
func (ts *TaskManager) getEntryPC(callerSkip int, checkGlobal bool) uintptr {
	// !internal.IgnorePC =>TRUE
	var pcs [1]uintptr
	// skip [runtime.Callers, this function, this function's caller]
	if checkGlobal && ts.isGlobal {
		runtime.Callers(callerSkip+1, pcs[:])
	} else {
		runtime.Callers(callerSkip, pcs[:])
	}
	return pcs[0]
}

// log 日志记录写入, 支持自定义 slog.Record.PC
func (ts *TaskManager) log(pc uintptr, level slog.Level, msg string, args ...any) {
	if !GetLogger().Handler().Enabled(ts.logCtx, level) {
		return
	}
	args = append(args, slog.String("fromTaskManager", ts.name))
	if pc == 0 {
		GetLogger().Log(ts.logCtx, level, msg, args...)
		return
	}
	record := slog.NewRecord(time.Now(), level, msg, pc)
	record.Add(args...)
	GetLogger().Handler().Handle(ts.logCtx, record)
}

// logFormat 日志记录写入, 支持自定义 slog.Record.PC
func (ts *TaskManager) logFormat(pc uintptr, level slog.Level, format string, args ...any) {
	if !GetLogger().Handler().Enabled(ts.logCtx, level) {
		return
	}
	if pc == 0 {
		GetLogger().Log(ts.logCtx, level, fmt.Sprintf(format, args...), slog.String("fromTaskManager", ts.name))
		return
	}
	record := slog.NewRecord(time.Now(), level, fmt.Sprintf(format, args...), pc)
	record.AddAttrs(slog.String("fromTaskManager", ts.name))
	GetLogger().Handler().Handle(ts.logCtx, record)
}
